package cn.sunline.dreamvalue.service.impl;

import cn.sunline.dreamvalue.dto.*;
import cn.sunline.dreamvalue.entity.CustDreamValFlow;
import cn.sunline.dreamvalue.service.HbaseService;
import cn.sunline.dreamvalue.utils.HbaseOprateUtil;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.util.*;

/**
 * @author 11292
 * @title: HbaseServiceImpl
 * @projectName dream-value
 * @description: hbase操作实现类
 * @date 2020/5/1123:13
 */
@Service
public class HbaseServiceImpl implements HbaseService {

    public static final Logger log = LoggerFactory.getLogger(HbaseServiceImpl.class);

    @Autowired
    private HbaseOprateUtil hbaseOprateUtil;

    //客户证件映射表
    public final String certNoMap = "t99_cert_no_map";

    //客户号映射表
    public final String customerIdMap = "t99_customer_id_map";

    //客户梦想账户表
    public final String custDreamAcct = "t04_cust_dream_acct";

    //客户梦想值流水表
    public final String custDreamValFlow = "t05_cust_dream_val_flow";

    //客户梦想任务完成进度表
    public final String custDreamTaskCmpltProg = "t05_cust_dream_task_cmplt_prog";
    
    /**
     　* @description: 查询证件号映射
     　* @param [certNo]
     　* @return java.util.Map<java.lang.String,java.lang.String>
     　* @author 11292
     　* @date 2020/5/12 10:05
     */
    @Override
    public Map<String, String> queryCertNoRel(String certNo) throws IOException {
        Map<String, List<String>> infos = new HashMap<String, List<String>>();
        List<String> columns = new ArrayList<>();
        columns.add("a0");
        columns.add("a1");
        columns.add("a2");
        infos.put("bi",columns);
        Map<String,String> result = hbaseOprateUtil.query(certNoMap,certNo,infos);
        return result;
    }

    /**
     　* @description: 查询客户号映射
     　* @param [certNo, customerId, dreamAcctNo]
     　* @return java.lang.Boolean
     　* @author 11292
     　* @date 2020/5/12 10:04
     */
    @Override
    public Map<String, String> queryCustomerIdRel(String customerId) throws IOException {
        Map<String, List<String>> infos = new HashMap<String, List<String>>();
        List<String> columns = new ArrayList<>();
        columns.add("a0");
        columns.add("a1");
        columns.add("a2");
        infos.put("bi",columns);
        Map<String,String> result = hbaseOprateUtil.query(customerIdMap,customerId,infos);
        return result;
    }

    /**
     　* @description: 插入证件号映射
     　* @param [certNo, customerId, dreamAcctNo]
     　* @return java.lang.Boolean
     　* @author 11292
     　* @date 2020/5/12 10:04
     */
    @Override
    public Boolean insertCertNoRel(String certNo, String customerId, String dreamAcctNo) throws IOException {
        Map<String,String> cols = new HashMap<>();
        cols.put("bi:a0",certNo);
        cols.put("bi:a1",customerId);
        cols.put("bi:a2",dreamAcctNo);
        String result = hbaseOprateUtil.update(certNoMap,certNo,cols);
        if(null != result && result.equals("success")){
            return true;
        }
        return false;
    }

    /**
     　* @description: 插入客户号映射
     　* @param [customerId, certNo, dreamAcctNo]
     　* @return java.lang.Boolean
     　* @author 11292
     　* @date 2020/5/12 10:03
     */
    @Override
    public Boolean insertCustomerIdRel(String customerId, String certNo, String dreamAcctNo) throws IOException {
        Map<String,String> cols = new HashMap<>();
        cols.put("bi:a0",customerId);
        cols.put("bi:a1",certNo);
        cols.put("bi:a2",dreamAcctNo);
        String result = hbaseOprateUtil.update(customerIdMap,customerId,cols);
        if(null != result && result.equals("success")){
            return true;
        }
        return false;
    }

    /**
     　* @description: 插入客户梦想账户
     　* @param [dreamAcctNo, isStartUp, date, time]
     　* @return java.lang.Boolean
     　* @author 11292
     　* @date 2020/5/12 10:02
     */
    @Override
    public Boolean insertCustDreamAcct(String dreamAcctNo, String isStartUp, String date, String time) throws IOException {
        Map<String,String> cols = new HashMap<>();
        cols.put("bi:a0",dreamAcctNo);
        cols.put("bi:a1",isStartUp);
        cols.put("bi:a2",date);
        cols.put("bi:a3",time);
        cols.put("bi:a4",date+time);
        String result = hbaseOprateUtil.update(custDreamAcct,dreamAcctNo,cols);
        if(null != result && result.equals("success")){
            return true;
        }
        return false;
    }

    /**
     　* @description: 删除客户号映射
     　* @param [customerNo]
     　* @return void
     　* @author 11292
     　* @date 2020/5/12 10:01
     */
    @Override
    public void deleteCustomerIdRel(String customerNo) throws IOException {
        hbaseOprateUtil.delete(customerIdMap,Arrays.asList(customerNo));
    }

    /**
     　* @description: 删除证件号映射
     　* @param [certNo]
     　* @return void
     　* @author 11292
     　* @date 2020/5/12 10:01
     */
    @Override
    public void deleteCertNoRel(String certNo) throws IOException {
        hbaseOprateUtil.delete(certNoMap,Arrays.asList(certNo));
    }

    /**
     　* @description: 查询梦想账户
     　* @param [dreamValueNo]
     　* @return java.util.Map<java.lang.String,java.lang.String>
     　* @author 11292
     　* @date 2020/5/12 10:00
     */
    @Override
    public Map<String, String> queryDreamAcct(String dreamValueNo) throws IOException {
        Map<String, List<String>> infos = new HashMap<String, List<String>>();
        List<String> columns = new ArrayList<>();
        columns.add("a0");
        columns.add("a1");
        columns.add("a2");
        columns.add("a3");
        columns.add("a4");
        infos.put("bi",columns);
        Map<String,String> result = hbaseOprateUtil.query(custDreamAcct,dreamValueNo,infos);
        return result;
    }

    /**
     　* @description: 查询梦想值流水
     　* @param [input]
     　* @return cn.sunline.dreamvalue.dto.DreamValueDetailOutDTO
     　* @author 11292
     　* @date 2020/5/12 10:00
     */
    @Override
    public DreamValueDetailOutDTO queryDreamValFlow(DreamValueDetailInDTO input) throws Exception {
        return queryDreamValFlowByRowKey(input);
    }

    /**
     * 通过filter查询梦想值流水
     * @param input
     * @return
     */
    private DreamValueDetailOutDTO queryDreamValFlowByFilter(DreamValueDetailInDTO input) throws Exception{
        DreamValueDetailOutDTO data = new DreamValueDetailOutDTO();
        List<DreamValueDetailDTO> list = new ArrayList<DreamValueDetailDTO>();

        ResultScanner resultScanner = hbaseOprateUtil.queryDreamValuFlows(custDreamValFlow,input);
        data.setDreamValueNo(input.getDreamValueNo());
        for(Result result : resultScanner){
            List<Cell> cells = result.listCells();
            DreamValueDetailDTO dto = transRowformCells(cells);
            list.add(dto);
        }
        if(null != list && !list.isEmpty()){
            data.setList(list);
        }

        return data;
    }


    /**
     * 通过rowkey的形式查询梦想值流水
     * @param infos
     * @return
     * @throws Exception
     */
    private DreamValueDetailOutDTO queryDreamValFlowByRowKey(DreamValueDetailInDTO infos) throws Exception{
        if(null == infos.getSize() || infos.getSize() < 1){
            infos.setSize(10);
        }
        DreamValueDetailOutDTO data = new DreamValueDetailOutDTO();
        data.setDreamValueNo(infos.getDreamValueNo());
        List<DreamValueDetailDTO> list = new ArrayList<DreamValueDetailDTO>();
        DateFormat df = new SimpleDateFormat("yyyyMMdd");
        String startRowKey = infos.getDreamValueNo()+infos.getType();
        String endRowKey = infos.getDreamValueNo() + infos.getType();
        if(StringUtils.isNotEmpty(infos.getStartTime())){
            startRowKey += infos.getStartTime();
        }else {
            Date date = new Date(0);
            startRowKey += df.format(date);
        }
        if(StringUtils.isNotEmpty(infos.getEndTime())){
            endRowKey += infos.getEndTime();
        }else {
            Date date = new Date(System.currentTimeMillis()+1000*60*60*24);
            endRowKey += df.format(date);
        }
        if(StringUtils.isNotEmpty(infos.getLastNo())){
            endRowKey += infos.getLastNo();
        }

        log.info("===============================>>>>>>>>startRowKey is {}",startRowKey);
        log.info("===============================>>>>>>>>endRowKey is {}",endRowKey);
        //定义返回字段
        Map<String,List<String>> colums = new HashMap<String,List<String>>();
        List<String> columnList = Arrays.asList("a2","a3","a5","a6","a8","a9");
        colums.put("bi",columnList);
        List<Map<String,String>> result = hbaseOprateUtil.queryDesc(custDreamValFlow,endRowKey,startRowKey,colums,infos.getSize());

        if(null != result && !result.isEmpty()){
            for(int i = 0; i < result.size(); i ++){
                Map<String,String> res = result.get(i);
                DreamValueDetailDTO dto= new DreamValueDetailDTO();
                if(null != res && !res.isEmpty()){
                    dto.setTransNo(res.get("bi:a6"));
                    dto.setTransDesc(res.get("bi:a8"));
                    dto.setDate(res.get("bi:a2"));
                    dto.setTime(res.get("bi:a3"));
                    dto.setDreamValue(res.get("bi:a9"));
                    dto.setFlowNo(res.get("bi:a5"));
                }
                if(StringUtils.isNotEmpty(infos.getLastNo()) && infos.getSize() < result.size() && i == result.size() -1){
                    //如果首页，则踢掉最后一条数据
                    break;
                }
                if(!list.contains(dto)){
                    list.add(dto);
                }
            }
        }

        if(null != list && !list.isEmpty()){
            data.setList(list);
        }
        return data;
    }


    /**
    　* @description: 往hbase梦想值流水记录表中插入记录
    　* @param
    　* @return
    　* @author zjp
    　* @date 2020/5/12 16:10
    */
    @Override
    public Boolean insertCustDreamValFlow(CustDreamValFlow flow) throws IOException {
        String rowKey = flow.getDreamAcctNo() + flow.getDreamAcctType() + flow.getTxDate()+flow.getTxTime()+ flow.getFlowNo();
        DateFormat df = new SimpleDateFormat("yyyyMMddHHmmss");
        Map<String,String> cols = new HashMap<>();
        cols.put("bi:a0",flow.getDreamAcctNo());
        cols.put("bi:a1",flow.getDreamAcctType());
        cols.put("bi:a2",flow.getTxDate());
        cols.put("bi:a3",flow.getTxTime());
        cols.put("bi:a4",flow.getOperType());
        cols.put("bi:a5",flow.getFlowNo());
        cols.put("bi:a6",flow.getTxSeqNo());
        cols.put("bi:a7",df.format(flow.getTxDateTime()));
        cols.put("bi:a8",flow.getTxDesc());
        cols.put("bi:a9",flow.getDreamVal().toString());
        cols.put("bi:b0",df.format(flow.getInsertTime()));
        String result = hbaseOprateUtil.update(custDreamValFlow,rowKey,cols);
        if(null != result && result.equals("success")){
            return true;
        }
        return false;
    }


    /**
     　* @description: 转换查询结果
     　* @param [cells]
     　* @return cn.sunline.dreamvalue.dto.DreamValueDetailDTO
     　* @author 11292
     　* @date 2020/5/12 14:58
     */
    private  DreamValueDetailDTO transRowformCells(List<Cell> cells) throws UnsupportedEncodingException {
        Map<String,String> res = new HashMap<String,String>();
        if(null == cells || cells.size() == 0){
            return null;
        }
        for (Cell cell : cells) {
            res.put(new String(CellUtil.cloneFamily(cell),"utf-8")+":"+ new String(CellUtil.cloneQualifier(cell),"utf-8"),new String(CellUtil.cloneValue(cell),"utf-8"));
        }
        DreamValueDetailDTO dto= new DreamValueDetailDTO();
        if(null != res && !res.isEmpty()){
            dto.setTransNo(res.get("bi:a6"));
            dto.setTransDesc(res.get("bi:a8"));
            dto.setDate(res.get("bi:a2"));
            dto.setTime(res.get("bi:a3"));
            dto.setDreamValue(res.get("bi:a9"));
            dto.setFlowNo(res.get("bi:a5"));
        }
        return dto;
    }

    /**
     　* @description: 查询客户完成度
     　* @param [input]
     　* @return cn.sunline.dreamvalue.dto.DreamValueTaskOutDTO
     　* @author 82345
     　* @date 2020/5/14 12:14
     */
    @Override
    public DreamValueTaskOutDTO querySchedule(DreamAcctOpenDTO input) throws Exception {

        ResultScanner resultScanner = hbaseOprateUtil.queryDreamValuTask(custDreamTaskCmpltProg,input);
        DreamValueTaskOutDTO data = new DreamValueTaskOutDTO();
        data.setDreamValueNo(input.getDreamValueNo());
        List<DreamValueTaskDTO> list = new ArrayList<DreamValueTaskDTO>();
        for(Result result : resultScanner){
            List<Cell> cells = result.listCells();
            DreamValueTaskDTO dto = transRowformCellsTask(cells);
            list.add(dto);
        }
        if(null != list && !list.isEmpty()){
            data.setList(list);
        }
        return data;
    }
    /**
     　* @description: 转换查询结果
     　* @param [cells]
     　* @return cn.sunline.dreamvalue.dto.DreamValueTaskDTO
     　* @author 82345
     　* @date 2020/5/14 12:13
     */
    private  DreamValueTaskDTO transRowformCellsTask(List<Cell> cells) throws UnsupportedEncodingException {

        Map<String,String> res = new HashMap<String,String>();
        if(null == cells || cells.size() == 0){
            return null;
        }
        for (Cell cell : cells) {
            res.put(new String(CellUtil.cloneFamily(cell),"utf-8")+":"+ new String(CellUtil.cloneQualifier(cell),"utf-8"),new String(CellUtil.cloneValue(cell),"utf-8"));
        }
        DreamValueTaskDTO dto= new DreamValueTaskDTO();
        if(null != res && !res.isEmpty()){
            dto.setTaskId(res.get("bi:a3"));
            dto.setTaskStatus(res.get("bi:a7"));
            dto.setReachTime(res.get("bi:a8"));
            dto.setReachNum(res.get("bi:a9"));
        }
        return dto;
    }
}
